package main.test.GraphFramesAPI

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext
import org.graphframes.GraphFrame

object SaveLoadGraphFrames {

  def main(args: Array[String]): Unit = {
    //  def constructGraph: GraphFrame = {
    //    The following example demonstrates how
    //    to create a GraphFrame from vertex and edge DataFrame
    val sparkConf = new SparkConf()
    sparkConf.setAppName("createGraph").setMaster("local[*]")
    val sc = new SparkContext(sparkConf)
    val sqlContext = new SQLContext(sc)
    val v = sqlContext.createDataFrame(List(
      ("a", "Alice", 34),
      ("b", "Bob", 36),
      ("c", "Charlie", 30),
      ("d", "David", 29),
      ("e", "Esther", 32),
      ("f", "Fanny", 36),
      ("g", "Gabby", 60)
    )).toDF("id", "name", "age")

    // Edge DataFrame
    val e = sqlContext.createDataFrame(List(
      ("a", "b", "friend"),
      ("b", "c", "follow"),
      ("c", "b", "follow"),
      ("f", "c", "follow"),
      ("e", "f", "follow"),
      ("e", "d", "friend"),
      ("d", "a", "friend"),
      ("a", "e", "friend")
    )).toDF("src", "dst", "relationship")


    // Create a GraphFrame
    val g = GraphFrame(v, e)

    /**
     * Since GraphFrames are built around DataFrames
     * they automatically support saving and loading to and from
     * the same set of data sources. Refer to the Spark SQL user guide
     * on datasources for more details.(
     * http://spark.apache.org/docs/latest/sql-programming-guide.html#data-sources)
     *
     * Below we show how to save and then load a graph
     */
    // Save vertices and edges as Parquet to some location.
//    g.vertices.write.parquet("hdfs://myLocation/vertices")
    g.vertices.write.parquet("./myLocation/vertices")
//    g.edges.write.parquet("hdfs://myLocation/edges")
    g.edges.write.parquet("./myLocation/edges")

    // Load the vertices and edges back.
    val sameV = sqlContext.read.parquet("./myLocation/vertices")
    val sameE = sqlContext.read.parquet("./myLocation/edges")

    // Create an identical GraphFrame.
    val sameG = GraphFrame(sameV, sameE)






  }

}
